"""step5_3_prepare_d_view

目的：
    在物理 FEE 表上追加并填充需要的融合 / 计算列（来自 MDTRT / SETL / 派生逻辑），随后创建视图 d。
    视图 m 只展示 j_drop IS NULL 的有效数据（无效行 j_drop='1' 不出现在视图中），并按 env_structure.xlsx 中的别名(core) 输出列名。
    不做字典汉化 / _MAP 字段。
    不生成 j_refund；保留 j_rand；j_drop= '1' 表示无效，正常为 NULL。

步骤概要：
    1. 读取 env_structure.xlsx (获取 SETL / MDTRT / FEE 字段列表及 core 别名)。注意：全都转换为大写
    2. 解析环境变量：
        2.1. 表名 SETL_TBL_NAME / MDTRT_TBL_NAME / FEE_TBL_NAME
        2.2. 医院简称 HSP_ABBR 字典
    3. 为 FEE 表新增 SETL 的派生字段：HSP_ABBR, J_ISIN, J_HAS_B, J_HSP_LV, IN_DAY, OUT_DAY, PERIOD, J_YEAROLD, J_DAYOLD_IN, J_DAYOLD_OUT, J_YEAR, GNR_C, GNR_B, HIFP_PAY
        3.1. 如果已经存在的，则先TRUNCATE；不存在的则新建列
        3.2. 将数据从 SETL 表中更新到 FEE 表
    4. 为 FEE 表新增计算字段：ITEM_DAY, ITEM_HOUR, ITEM_J_CODE, ITEM_J_NAME, ITEM_STD_CODE, ITEM_STD_NAME。统一删除然后再重新创建
        4.1. ITEM_DAY： 费用发生日期，基于 ITEM_TIME 截断时间部分
        4.2. ITEM_HOUR： 费用发生小时，基于 ITEM_TIME 截取小时部分
        4.3. ITEM_J_CODE / ITEM_J_NAME： 费用项目编码/名称，基于多个 ITEM_CODE 的组合
        4.4. ITEM_STD_CODE / ITEM_STD_NAME： 费用项目标准编码/名称，基于 ITEM_CODE 映射，映射规则来自于 step5_3_create_d 的规则
    5. 创建索引：setl_id, item_j_code, item_time, item_type, hsp_abbr, psn_no, item_std_code
    6. 添加主键：D0_ROWID, ITEM_TIME_SEQ，两列构造逻辑一样，构造规则来自于 step5_3_create_d 的规则。其中 DO_ROWID 设为主键，ITEM_TIME_SEQ 设为唯一索引
    7. 处理冲减
        7.1. 添加联合索引 ITEM_TIME_SEQ, SETL_ID
        7.2. 新增 J_Q, J_P, J_R 三列 NUMBER、新增 IS_J_REDUCE 一列 VARCHAR2(1)
        7.3. 计算冲减
        7.4. 新增 J_B, IS_J_REDUCE 索引
        7.5. 添加开关列，并且创建索引
    8. 创建视图 d

运行：
   python STEP5干净表/step5_3_prepare_d_view.py
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from typing import List, Dict
import argparse
import pandas as pd
from sqlalchemy import text
from sqlalchemy.engine import Engine
import time

ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
if ROOT_DIR not in sys.path:
    sys.path.append(ROOT_DIR)

from config import create_db_engine, load_env_tbl_name, load_env_structure_setl, load_env_db

# 元数据文件与环境变量键名常量
STRUCTURE_FILE = 'env.xlsx'
STRUCTURE_SHEET = 'env_structure'

# 获得所有表名的环境变量键名
SETL_TBL_NAME, MDTRT_TBL_NAME, FEE_TBL_NAME, DX_TBL_NAME, TX_TBL_NAME = load_env_tbl_name()

SETL_ENV = SETL_TBL_NAME
MDTRT_ENV = MDTRT_TBL_NAME
FEE_ENV = FEE_TBL_NAME

# 从 env 获得 DB_USER
DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_SERVICE_NAME, DB_DATABASE = load_env_db()

# 时间函数
t0 = time.time()
def elapsed() -> str:
    """返回自脚本启动以来的耗时 (HH:MM:SS)。
    用于将原先的 [INFO]/[START]/[DONE] 等等级标签替换成实时耗时。
    """
    
    # 实际时间，yyy-mm-dd hh:mm:ss
    timeStr = time.strftime("%H:%M:%S", time.localtime())
    
    # 已消耗时间 XX hours XX minutes XX.XX seconds
    delta = int(time.time() - t0)
    if delta < 60:
        return f"{timeStr} (+ {delta} sec)"
    elif delta < 3600:
        m, s = divmod(delta, 60)
        return f"{timeStr} (+ {m} min {s} sec)"
    elif delta < 86400:
        h, rem = divmod(delta, 3600)
        m, s = divmod(rem, 60)
        return f"{timeStr} (+ {h} hour {m} min {s} sec)"
    else:
        d, rem = divmod(delta, 86400)
        h, rem = divmod(rem, 3600)
        m, s = divmod(rem, 60)
        return f"{timeStr} (+ {d} day {h} hour {m} min {s} sec)"

# ------------------------------
#  会话与性能参数
# ------------------------------
def get_parallel_degree() -> int:
    try:
        return max(1, 8)
    except Exception:
        return 8

def setup_oracle_parallel_session(engine: Engine, degree: int):
    """为本次会话开启并行 DML/QUERY，设置合理的并行度。

    可通过环境变量 ORA_PARALLEL_DEGREE 控制并行度（默认 8）。
    """
    with engine.begin() as conn:
        try:
            conn.execute(text("ALTER SESSION ENABLE PARALLEL DML"))
            conn.execute(text(f"ALTER SESSION FORCE PARALLEL QUERY PARALLEL {degree}"))
            # 对临时/批量对象建表/建索引时减少日志
            # 注：NOLOGGING 在具体 DDL 中指定，这里不改数据库级参数
            print(f"[{elapsed()}] 已启用并行会话：PARALLEL DML/QUERY，DEGREE={degree}")
        except Exception as e:
            print(f"[{elapsed()}] 并行会话设置失败（继续执行）：{e}")

# ------------------------------
#  读取配置与环境
# ------------------------------
def read_env_structure(file_path: str = STRUCTURE_FILE) -> pd.DataFrame:
    # 读取 env_structure.xlsx，返回 DataFrame
    df_raw = pd.read_excel(file_path, sheet_name=STRUCTURE_SHEET)
    needed = ['tbl_name', 'field_code', 'field_name', 'core']
    for c in needed:
            if c not in df_raw.columns:
                    raise ValueError(f'env_structure.xlsx 缺少列: {c}')

    df = df_raw[needed].copy()
    # 标准化
    df['tbl_name'] = df['tbl_name'].astype(str).str.strip().str.upper()
    df['field_code'] = df['field_code'].astype(str).str.strip().str.upper()
    df['core'] = df['core'].fillna('').astype(str).str.strip().str.upper()
    df['field_name'] = df['field_name'].fillna('').astype(str).str.strip()
    print(f"[{elapsed()}] 读取结构表，行数={len(df)}")
    return df

def resolve_table_names() -> tuple[str, str, str]:
    # 读取环境变量中的核心表名 SETL, MDTRT, FEE
    setl = SETL_TBL_NAME
    mdtrt = MDTRT_TBL_NAME
    fee = FEE_TBL_NAME
    print(f"[{elapsed()}] 解析环境变量: {SETL_ENV}={setl}, {MDTRT_ENV}={mdtrt}, {FEE_ENV}={fee}")
    return setl, mdtrt, fee

def resolve_cores(struct_df: pd.DataFrame, tbl_name: str) -> Dict[str, str]:
    # 从结构表中解析指定表的 core -> field_code 映射
    df = struct_df[struct_df['tbl_name'] == tbl_name.upper()]
    core_map = {}
    for _, row in df.iterrows():
        core = row['core']
        field_code = row['field_code']
        if core:
            core_map[core] = field_code
    print(f"[{elapsed()}] 解析 {tbl_name} 核心列映射，共 {len(core_map)} 列")
    return core_map

# ------------------------------
#  FEE 表新增 SETL 表的派生字段
# ------------------------------
def _column_exists(engine: Engine, table: str, col: str) -> bool:
    sql = text("SELECT 1 FROM user_tab_columns WHERE table_name=:t AND column_name=:c")
    with engine.begin() as conn:
        return conn.execute(sql, {'t': table.upper(), 'c': col.upper()}).fetchone() is not None

def _truncate_columns(engine: Engine, table: str, cols: List[str]):
    ddl = f"UPDATE {table} SET {', '.join([f'{col} = NULL' for col in cols])}"
    with engine.begin() as conn:
        conn.execute(text(ddl))

def _add_column(engine: Engine, table: str, col: str, dtype: str):
    ddl = f"ALTER TABLE {table} ADD ({col} {dtype})"
    with engine.begin() as conn:
        try:
            conn.execute(text(ddl))
        except Exception as e:
            msg = str(e)
            # ORA-01430: column being added already exists in table / ORA-00955 object already exists
            if 'ORA-01430' in msg or 'already exists' in msg or 'ORA-00955' in msg:
                print(f"[{elapsed()}] {col} 已存在 (SKIP)")
            else:
                raise

def _get_existing_columns(engine: Engine, table: str) -> set[str]:
    sql = text("SELECT column_name FROM user_tab_columns WHERE table_name=:t")
    with engine.begin() as conn:
        rows = conn.execute(sql, { 't': table.upper() }).fetchall()
    return {r[0].upper() for r in rows}

def _add_columns(engine: Engine, table: str, columns: List[tuple[str, str]]):
    """一次性新增多列: columns=[(col, dtype), ...]"""
    if not columns:
        return
    parts = [f"{col} {dtype}" for col, dtype in columns]
    ddl = f"ALTER TABLE {table} ADD (" + ", ".join(parts) + ")"
    with engine.begin() as conn:
        try:
            conn.execute(text(ddl))
        except Exception as e:
            # 如果部分已存在或其他 DDL 冲突，逐列回退尝试
            print(f"[{elapsed()}] 批量新增列失败，逐列回退：{e}")
            for col, dtype in columns:
                _add_column(engine, table, col, dtype)

def _infer_type(col: str) -> str:
    u = col.upper()
    if any(k in u for k in ['TIME','DATE','ITEM_DAY','IN_DAY','OUT_DAY']):
        return 'DATE'
    elif any(k in u for k in ['AMT','FEE','SUM','COST','AGE','PERIOD','YEAR','DAY','GNR_C','GNR_B','HIFP_PAY']):
        return 'NUMBER'
    return 'VARCHAR2(400)'

def prepare_fee_with_setl(
    engine: Engine, 
    feeTblName: str, setlTblName: str,
    feeCoresMap: Dict[str, str], setlCoresMap: Dict[str, str]
):
    # 为 FEE 表新增 SETL 的派生字段
    setl_derived_core = [
        'HSP_ABBR', 'J_ISIN', 'J_HAS_B', 'J_HSP_LV', 'IN_DAY', 'OUT_DAY', 'PERIOD', 'J_GEND',
        'J_YEAROLD', 'J_DAYOLD_IN', 'J_DAYOLD_OUT', 'J_YEAR',
    ]
    # 表名, field_code, core(as_name)
    setl_fields = [('setl', core, core) for core in setl_derived_core]
    # 为 FEE 表新增 SETL 的固有字段
    setl_inherent_cores = [
        'MDTRT_ID', 'PSN_NO', 'IN_DEPT', 'OUT_DEPT', 'IN_TIME', 'OUT_TIME',
        'IN_DEPT', 'OUT_DEPT', 'AGE', 'MED_TYPE', 'INSUTYPE'
    ]
    setl_fields += [('setl', setlCoresMap[core], core) for core in setl_inherent_cores if core in setlCoresMap]
    # 批量新增缺失列
    existing = _get_existing_columns(engine, feeTblName)
    missing_defs: List[tuple[str,str]] = []
    exists_fields: List[str] = []
    for tblName, field, core in setl_fields:
        if field.upper() in existing:
            exists_fields.append(field)
        else:
            dtype = _infer_type(field)
            missing_defs.append((field, dtype))
    if missing_defs:
        _add_columns(engine, feeTblName, missing_defs)
        print(f"[{elapsed()}] 批量新增列 {len(missing_defs)} 个: {', '.join([c for c,_ in missing_defs])}")
    print(f"[{elapsed()}] FEE 表格式调整，SETL 派生列：总 {len(setl_fields)}，已存在 {len(exists_fields)}，新增 {len(missing_defs)}")
    
    # 获取 fee 表中的 setl_id 列名；setl 表的 setl_id 列名
    fee_setlId = feeCoresMap['SETL_ID']
    setl_setlId = setlCoresMap['SETL_ID']
    # 制作 joinKeys
    joinKeys = ', '.join([f"f.{field} = s.{field}" for tblName, field, core in setl_fields])
    deg = get_parallel_degree()
    update_sql = text(f"""
        MERGE /*+ parallel(f {deg}) parallel(s {deg}) use_hash(s) */ INTO {feeTblName} f
        USING {setlTblName} s
        ON (f.{fee_setlId} = s.{setl_setlId})
        WHEN MATCHED THEN UPDATE SET
            {joinKeys} 
    """)
    with engine.begin() as conn:
        conn.execute(update_sql)
    print(f"[{elapsed()}] FEE 表更新 SETL 派生列完成")

# ------------------------------
#  新增日期计算字段
# ------------------------------
def prepare_fee_with_dates(engine: Engine, feeTblName: str, feeCoresMap: Dict[str, str]):
    # 为 FEE 表新增计算字段：ITEM_DAY, ITEM_HOUR, ITEM_J_CODE, ITEM_J_NAME, ITEM_STD_CODE, ITEM_STD_NAME
    date_fields = [
        'ITEM_DAY', 'ITEM_HOUR'
    ]
    # 根据现有字段计算
    fee_itemTime = feeCoresMap['ITEM_TIME']
    # 批量新增缺失列
    existing = _get_existing_columns(engine, feeTblName)
    missing_defs: List[tuple[str,str]] = []
    exists_fields: List[str] = []
    for field in date_fields:
        if field.upper() in existing:
            exists_fields.append(field)
        else:
            dtype = _infer_type(field)
            missing_defs.append((field, dtype))
    if missing_defs:
        _add_columns(engine, feeTblName, missing_defs)
        print(f"[{elapsed()}] 批量新增日期列 {len(missing_defs)} 个: {', '.join([c for c,_ in missing_defs])}")
    print(f"[{elapsed()}] FEE 表格式调整，日期列：总 {len(date_fields)}，已存在 {len(exists_fields)}，新增 {len(missing_defs)}")
    
    # 更新
    deg = get_parallel_degree()
    sql = text(f"""
        UPDATE /*+ parallel({feeTblName} {deg}) */ {feeTblName}
        SET ITEM_DAY = TRUNC({fee_itemTime}),
            ITEM_HOUR = TO_NUMBER(TO_CHAR({fee_itemTime}, 'HH24'))
        WHERE {fee_itemTime} IS NOT NULL
          AND (ITEM_DAY IS NULL OR ITEM_HOUR IS NULL)
    """)
    with engine.begin() as conn:
        conn.execute(sql)
    print(f"[{elapsed()}] FEE 表更新 日期计算列完成")

# ------------------------------
#  为 FEE 表新增 标准编码相关的字段
# ------------------------------
def prepare_fee_with_std_code(engine: Engine, feeTblName: str, feeCoresMap: Dict[str, str]):
    """标准化费用项目编码/名称/类别 (不做兜底)。

    字段含义：
      - ITEM_STD_CODE : 第一次命中规则的原始来源编码 (来自 ITEM_CODE 或 ITEM_MED_CODE)。
      - ITEM_STD_NAME : 对应名称 (来自 ITEM_NAME)。
      - ITEM_TYPE     : 标准费用类别 (仅规则提供分类时写入；不覆盖已写入的类别)。

    规则执行：
      1. 先 LV0(高优先) 再 LV1；同级按列表顺序。
      2. 每条规则先尝试 ITEM_CODE，再尝试 ITEM_MED_CODE；谁先命中谁锁定。
      3. 只在 ITEM_STD_CODE 为空时允许赋值 (保持首次命中)。
      4. ITEM_TYPE 只有在规则带有分类且当前列仍为空时才写入。
      5. 不进行任何兜底拼接逻辑；未命中保持 NULL。
    """
    # 取得物理列名
    code_primary = feeCoresMap['ITEM_CODE']        # hilist_code
    code_secondary = feeCoresMap['ITEM_MED_CODE']  # med_list_codg
    fee_itemName = feeCoresMap['ITEM_NAME']  # hilist_name (可能缺失)

    # 准备目标列（批量新增缺失列）
    target_cols = ['ITEM_STD_CODE', 'ITEM_STD_NAME', 'ITEM_TYPE']
    existing = _get_existing_columns(engine, feeTblName)
    missing_defs: List[tuple[str,str]] = []
    for col in target_cols:
        if col.upper() not in existing:
            if col == 'ITEM_TYPE':
                missing_defs.append((col, 'VARCHAR2(200)'))
            elif col == 'ITEM_STD_CODE':
                missing_defs.append((col, 'VARCHAR2(200)'))
            else:  # ITEM_STD_NAME
                missing_defs.append((col, 'VARCHAR2(1000)'))
    if missing_defs:
        _add_columns(engine, feeTblName, missing_defs)
        print(f"[{elapsed()}] 批量新增标准化列 {len(missing_defs)} 个: {', '.join([c for c,_ in missing_defs])}")
    # 重要：不再清空全表，改为只对 NULL 进行填充，支持增量 & 多次运行
    print(f"[{elapsed()}] 标准化列已就绪（保持已写入数据，不执行全表清空）")

    # (规则名, 模板pattern, 费用类别)  —— 与原 create_d 逻辑的分类对应
    lv0_rules = [
        ('T开头10位中药饮片费', "{field} LIKE 'T%' AND LENGTH({field}) = 10 AND REGEXP_LIKE({field}, '^T[0-9]{9}$')", '中药饮片费'),
        ('T开头20位中药饮片费', "{field} LIKE 'T%' AND LENGTH({field}) = 20 AND REGEXP_LIKE({field}, '^T[0-9]{19}$')", '中药饮片费'),
        ('J44开头16位', "{field} LIKE 'J44%' AND LENGTH({field}) = 16 AND REGEXP_LIKE({field}, '^J44[0-9A-Z]{13}$')", None),
        ('0开头15位全数字', "{field} LIKE '0%' AND LENGTH({field}) = 15 AND REGEXP_LIKE({field}, '^[0-9]{15}$')", None),
        ('C开头20位', "{field} LIKE 'C%' AND LENGTH({field}) = 20 AND REGEXP_LIKE({field}, '^C[0-9]{19}$')", None),
        ('C开头27位', "{field} LIKE 'C%' AND LENGTH({field}) = 27 AND REGEXP_LIKE({field}, '^C[0-9]{26}$')", None),
        ('X开头23位西药费', "{field} LIKE 'X%' AND LENGTH({field}) = 23 AND REGEXP_LIKE({field}, '^X[A-Z][0-9]{2}[A-Z]{3}[0-9]{3}[A-Z][0-9]+$')", '西药费'),
        ('Z开头20位中成药费', "{field} LIKE 'Z%' AND LENGTH({field}) = 20 AND REGEXP_LIKE({field}, '^Z[A-Z][0-9]{2}[A-Z]{3}[0-9]+$')", '中成药费'),
        ('00或44开头25位', "({field} LIKE '00%' OR {field} LIKE '44%') AND LENGTH({field}) = 25 AND REGEXP_LIKE({field}, '^[0-9]{15}-[0-9A-Z]+$')", None),
        ('00或44开头26位', "({field} LIKE '00%' OR {field} LIKE '44%') AND LENGTH({field}) = 26 AND REGEXP_LIKE({field}, '^[0-9]{15}-[0-9A-Z]*[EFSN]$')", None),
        ('00或44开头27-30位', "({field} LIKE '00%' OR {field} LIKE '44%') AND LENGTH({field}) IN (27,28,29,30) AND REGEXP_LIKE({field}, '^[0-9]{15}-')", None),
    ]
    lv1_rules = [
        ('YP开头7位中药饮片费', "{field} LIKE 'YP%' AND LENGTH({field}) = 7", '中药饮片费'),
        ('9位全数字', "LENGTH({field}) = 9 AND REGEXP_LIKE({field}, '^[0-9]{9}$')", None),
        ('9位数字加EFSN', "LENGTH({field}) = 10 AND REGEXP_LIKE({field}, '^[0-9]{9}[EFSN]$')", None),
        ('9位数字其他变形', "LENGTH({field}) IN (11,12) AND REGEXP_LIKE({field}, '^[0-9]{9}')", None),
        ('9位数字加横杠斜杠', "LENGTH({field}) = 13 AND REGEXP_LIKE({field}, '^[0-9]{9}-[0-9]/[0-9]$')", None),
        ('C开头13位卫生材料费', "{field} LIKE 'C%' AND LENGTH({field}) = 13 AND REGEXP_LIKE({field}, '^C[0-9]{12}$')", '卫生材料费'),
        ('Z开头13位中成药费', "{field} LIKE 'Z-%' AND LENGTH({field}) = 13 AND REGEXP_LIKE({field}, '^Z-[A-Z][0-9]{2}[A-Z]{2}-[A-Z][0-9]{4}$')", '中成药费'),
        ('X开头17位西药费', "{field} LIKE 'X-%' AND LENGTH({field}) = 17 AND REGEXP_LIKE({field}, '^X-[A-Z][0-9]{2}[A-Z]{2}-[A-Z][0-9]{3}-[A-Z][0-9]{3}$')", '西药费'),
    ]

    # 使用单条 UPDATE + CASE WHEN，将所有规则按优先级串联，减少多次全表扫描
    # 构造 CASE WHEN（先 LV0 后 LV1；先 item_code 再 item_med_code；首次命中优先）
    code_when_clauses: List[str] = []
    type_when_clauses: List[str] = []

    def _append_rules_for_source(src_col: str, rules: List[tuple[str, str, str|None]]):
        for _, pattern_tpl, category in rules:
            cond = pattern_tpl.replace('{field}', src_col)
            code_when_clauses.append(f"WHEN {src_col} IS NOT NULL AND ({cond}) THEN {src_col}")
            if category:
                # 只有规则提供类别时，才生成对应的 WHEN 分支
                type_when_clauses.append(f"WHEN {src_col} IS NOT NULL AND ({cond}) THEN '{category}'")

    # 顺序：item_code 的 LV0→LV1，再到 item_med_code 的 LV0→LV1
    _append_rules_for_source(code_primary, lv0_rules)
    _append_rules_for_source(code_secondary, lv0_rules)
    _append_rules_for_source(code_primary, lv1_rules)
    _append_rules_for_source(code_secondary, lv1_rules)

    code_case_expr = "CASE " + " ".join(code_when_clauses) + " ELSE NULL END"
    type_case_expr = ("CASE " + " ".join(type_when_clauses) + " ELSE NULL END") if type_when_clauses else "NULL"

    deg = get_parallel_degree()
    update_all_sql = text(f"""
        UPDATE /*+ parallel({feeTblName} {deg}) */ {feeTblName}
        SET ITEM_STD_CODE = COALESCE(ITEM_STD_CODE, {code_case_expr}),
            ITEM_TYPE     = COALESCE(ITEM_TYPE, {type_case_expr})
        WHERE ITEM_STD_CODE IS NULL
          AND ( {code_primary} IS NOT NULL OR {code_secondary} IS NOT NULL )
    """)
    with engine.begin() as conn:
        res = conn.execute(update_all_sql)
        updated_cnt = res.rowcount or 0
    print(f"[{elapsed()}] 单条 UPDATE 完成标准编码与类别填充，更新行数 {updated_cnt} (未命中保持 NULL，无兜底)")

    # 额外更新 ITEM_TYPE
    fee_itemTypeRaw = feeCoresMap['ITEM_TYPE_RAW']
    update_type_sql = text(f"""
        UPDATE /*+ parallel({feeTblName} {deg}) */ {feeTblName}
        SET ITEM_TYPE = DECODE(
            {fee_itemTypeRaw},
            '01', '床位费',
            '02', '诊察费',
            '03', '检查费',
            '04', '化验费',
            '05', '治疗费',
            '06', '手术费',
            '07', '护理费',
            '08', '卫生材料费',
            '09', '西药费',
            '10', '中药饮片费',
            '11', '中成药费',
            '12', '一般诊疗费',
            '14', '其他费',
            NULL
        )
        WHERE ITEM_TYPE IS NULL
          AND {fee_itemTypeRaw} IS NOT NULL
    """)
    with engine.begin() as conn:
        res = conn.execute(update_type_sql)
        cnt = res.rowcount or 0
    print(f"[{elapsed()}] ITEM_TYPE 字典映射更新完成，累计更新 {cnt} 行")
    
    # 更新 item_std_name
    sql = text(f"""
        UPDATE /*+ parallel({feeTblName} {deg}) */ {feeTblName}
        SET item_std_name = {fee_itemName}
        WHERE item_std_name IS NULL
    """)
    with engine.begin() as conn:
        res = conn.execute(sql)
        cnt = res.rowcount or 0
    print(f"[{elapsed()}] 标准编码名称更新完成，累计更新 {cnt} 行")
    
    # 关联更新 serv 的 item_std_name
    merge_sql = text(f"""
        MERGE /*+ parallel(f {deg}) use_hash(s) */ INTO {feeTblName} f
        USING (
            SELECT /*+ parallel(8) */ item_j_code as item_std_code, item_j_name as item_std_name, item_type
            FROM STABLE_SERV
            WHERE item_j_code IS NOT NULL
        ) s
        ON (f.item_std_code = s.item_std_code)
        WHEN MATCHED THEN
            UPDATE SET 
                f.item_std_name = COALESCE(s.item_std_name, f.item_std_name),
                f.item_type = CASE 
                    WHEN s.item_type IS NOT NULL AND s.item_type != '' THEN s.item_type 
                    ELSE f.item_type 
                END
        """)
    with engine.begin() as conn:
        res = conn.execute(merge_sql)
        cnt = res.rowcount or 0
    print(f"[{elapsed()}] 关联 STABLE_SERV 更新标准编码名称完成，累计更新 {cnt} 行")

# ------------------------------
#  为 FEE 表新增 J项目编码相关的字段
# ------------------------------
def prepare_fee_with_j_code(
    engine: Engine, feeTblName: str, feeCoresMap: Dict[str, str]
):
    str_col = 'ITEM_J_CODE_STR'
    num_col = 'ITEM_J_CODE'

    def _get_col_meta(col: str) -> tuple[str | None, int | None]:
        meta_sql = text(
            """
            SELECT data_type, data_length
            FROM user_tab_columns
            WHERE table_name = :t AND column_name = :c
            """
        )
        with engine.begin() as conn:
            row = conn.execute(meta_sql, {"t": feeTblName.upper(), "c": col.upper()}).fetchone()
        if not row:
            return None, None
        return row[0], int(row[1]) if row[1] is not None else None

    # 确保字符串列存在，用于生成映射
    str_type, str_len = _get_col_meta(str_col)
    if str_type is None:
        # 若旧的 ITEM_J_CODE 存在且为字符型，则先改名保留字符串内容
        old_type, _ = _get_col_meta(num_col)
        if old_type and old_type.upper() != 'NUMBER':
            with engine.begin() as conn:
                conn.execute(text(f"ALTER TABLE {feeTblName} RENAME COLUMN {num_col} TO {str_col}"))
            print(f"[{elapsed()}] 将 {feeTblName}.ITEM_J_CODE 列重命名为 {str_col} 保存原始字符串")
        else:
            _add_column(engine, feeTblName, str_col, 'VARCHAR2(4000)')
            print(f"[{elapsed()}] 新增 {feeTblName}.{str_col} 列用于保存字符串编码")
    else:
        if str_len and str_len < 1000:
            with engine.begin() as conn:
                try:
                    conn.execute(text(f"ALTER TABLE {feeTblName} MODIFY ({str_col} VARCHAR2(4000))"))
                    print(f"[{elapsed()}] 将 {feeTblName}.{str_col} 列长度从 {str_len} 扩展到 4000")
                except Exception as e:
                    print(f"[{elapsed()}] 扩展 {str_col} 列长度失败（继续执行）：{e}")

    # 确保数字列存在
    num_type, _ = _get_col_meta(num_col)
    if num_type is None:
        _add_column(engine, feeTblName, num_col, 'NUMBER')
        print(f"[{elapsed()}] 新增 {feeTblName}.{num_col} 列 (NUMBER)")
    elif num_type.upper() != 'NUMBER':
        # 将旧值置空后再修改数据类型，避免转换失败
        with engine.begin() as conn:
            conn.execute(text(f"UPDATE {feeTblName} SET {num_col} = NULL"))
        with engine.begin() as conn:
            conn.execute(text(f"ALTER TABLE {feeTblName} MODIFY ({num_col} NUMBER)"))
        print(f"[{elapsed()}] 将 {feeTblName}.{num_col} 列转换为 NUMBER 类型")

    print(f"[{elapsed()}] FEE 表格式调整，确保 {str_col} 与 {num_col} 列可用")
    
    # 获得 item_code, item_hsp_code, p 对应的字段
    fee_itemCode = feeCoresMap['ITEM_CODE']
    fee_itemName = feeCoresMap['ITEM_NAME']
    fee_itemMedCode = feeCoresMap['ITEM_MED_CODE']
    fee_itemHspCode = feeCoresMap['ITEM_HSP_CODE']
    fee_itemHspName = feeCoresMap['ITEM_HSP_NAME']
    fee_p = feeCoresMap['P']
    
    item_j_code_expr = f"""
        SUBSTR(
            REPLACE(
                NVL({fee_itemCode},'') || '|' || 
                NVL({fee_itemName},'') || '|' ||
                NVL({fee_itemMedCode},'') || '|' || 
                NVL({fee_itemHspCode},'') || '|' || 
                NVL({fee_itemHspName},'') || '|' ||
                TO_CHAR(ROUND(NVL({fee_p},0)*1000, 0))
            , '''''', '_'),
            1, 1000
        )
    """.strip().replace("\n", " ")

    # 更新字符串列，后续用于映射生成
    deg = get_parallel_degree()
    sql_update_str = text(
        f"""
        UPDATE /*+ parallel({feeTblName} {deg}) */ {feeTblName}
           SET {str_col} = {item_j_code_expr}
        """
    )
    with engine.begin() as conn:
        conn.execute(sql_update_str)
    print(f"[{elapsed()}] FEE 表更新 {str_col} 字符串编码完成")

    # 创建临时映射表
    safe_tbl = ''.join(ch for ch in feeTblName.upper() if ch.isalnum())
    tmpTblName = f"TMP_{safe_tbl[:20]}_JCODE"[:30]
    pk_name = f"PK_{tmpTblName[:26]}"[:30]
    idx_name = f"IDX_{tmpTblName[:25]}_S"[:30]

    drop_tmp_sql = text(
        f"""
        BEGIN
            EXECUTE IMMEDIATE 'DROP TABLE {tmpTblName} PURGE';
        EXCEPTION
            WHEN OTHERS THEN NULL;
        END;
        """
    )
    with engine.begin() as conn:
        conn.execute(drop_tmp_sql)
    print(f"[{elapsed()}] 删除已存在的临时表 {tmpTblName} (如果有)")

    create_tmp_sql = text(
        f"""
        CREATE TABLE {tmpTblName}
        NOLOGGING
        PARALLEL {deg}
        AS
        SELECT
            str_val AS ITEM_J_CODE_STR,
            ROW_NUMBER() OVER (ORDER BY str_val) AS ITEM_J_CODE
        FROM (
            SELECT DISTINCT {str_col} AS str_val
            FROM {feeTblName}
            WHERE {str_col} IS NOT NULL
        )
        """
    )
    with engine.begin() as conn:
        conn.execute(create_tmp_sql)
    print(f"[{elapsed()}] 创建临时表 {tmpTblName} 完成")

    create_idx_num_sql = text(
        f"""
        ALTER TABLE {tmpTblName}
            ADD CONSTRAINT {pk_name} PRIMARY KEY (ITEM_J_CODE)
        """
    )
    with engine.begin() as conn:
        conn.execute(create_idx_num_sql)
    create_idx_str_sql = text(
        f"""
        CREATE UNIQUE INDEX {idx_name} ON {tmpTblName} (ITEM_J_CODE_STR)
        NOLOGGING
        PARALLEL {deg}
        """
    )
    with engine.begin() as conn:
        conn.execute(create_idx_str_sql)
    print(f"[{elapsed()}] 为临时表建立主键及唯一索引完成")

    merge_sql = text(
        f"""
        MERGE /*+ parallel(f {deg}) parallel(t {deg}) use_hash(t) */ INTO {feeTblName} f
        USING {tmpTblName} t
        ON (f.{str_col} = t.ITEM_J_CODE_STR)
        WHEN MATCHED THEN UPDATE SET f.{num_col} = t.ITEM_J_CODE
        """
    )
    with engine.begin() as conn:
        res = conn.execute(merge_sql)
        updated_cnt = res.rowcount or 0
    print(f"[{elapsed()}] FEE 表更新 {num_col} 数字编码完成，更新行数 {updated_cnt}")

    drop_tmp_final_sql = text(f"DROP TABLE {tmpTblName} PURGE")
    with engine.begin() as conn:
        conn.execute(drop_tmp_final_sql)
    print(f"[{elapsed()}] 清理临时表 {tmpTblName} 完成")

# ------------------------------
#  为 FEE 表增加时间顺序唯一码，并且设置主键与唯一索引
# ------------------------------
def _create_tmp_seq_table(engine: Engine, feeTblName: str, feeCoresMap: Dict[str, str]):
    tmpTblName = 'TMP_TIMESEQ'
    # 删除已存在的临时表
    drop_sql = text(f"""
        BEGIN
            EXECUTE IMMEDIATE 'DROP TABLE {tmpTblName} PURGE';
        EXCEPTION
            WHEN OTHERS THEN
                NULL; -- 忽略所有错误，包括表不存在的情况
        END;
    """)
    with engine.begin() as conn:
        conn.execute(drop_sql)
    print(f"[{elapsed()}] 删除已存在的临时表 {tmpTblName} (如果有)")
    
    # 获得 item_time 列名
    fee_itemTime = feeCoresMap['ITEM_TIME']
    
    # 创建一个临时的计算 item_time_seq 的表
    deg = get_parallel_degree()
    create_sql = text(f"""
        CREATE TABLE {tmpTblName}
        NOLOGGING
        PARALLEL {deg}
        AS
        SELECT /*+ parallel(d {deg}) */
            d.rowid as d0_rowid ,
            ROW_NUMBER() OVER (ORDER BY {fee_itemTime}, d.rowid) AS item_time_seq
        FROM {feeTblName} d
    """)
    with engine.begin() as conn:
        conn.execute(create_sql)
    print(f"[{elapsed()}] 创建临时表 {tmpTblName} 完成")
    
    # 为 d0_rowid 创建索引，方便更新
    idx_name = f"IDX_{tmpTblName}_D0ROWID"
    create_idx_sql = text(f"""
        CREATE INDEX {idx_name} ON {tmpTblName} (d0_rowid)
        NOLOGGING
        PARALLEL {deg}
    """)
    with engine.begin() as conn:
        conn.execute(create_idx_sql)
    print(f"[{elapsed()}] 创建索引 {idx_name} 完成")

def prepare_fee_with_time_seq(engine: Engine, feeTblName: str, feeCoresMap: Dict[str, str]):
    # 为 FEE 表新增 D0_ROWID, ITEM_TIME_SEQ 两列，并设置主键与唯一索引

    # 统一判断需要新增的列（无需清空，后续 MERGE 会覆盖全表数值）
    existing = _get_existing_columns(engine, feeTblName)
    time_cols = ['D0_ROWID']
    missing_defs: List[tuple[str, str]] = []
    exists_fields: List[str] = []
    for c in time_cols:
        if c.upper() in existing:
            exists_fields.append(c)
        else:
            missing_defs.append((c, 'NUMBER'))
    if missing_defs:
        _add_columns(engine, feeTblName, missing_defs)
        print(f"[{elapsed()}] 批量新增时间顺序列 {len(missing_defs)} 个: {', '.join([c for c,_ in missing_defs])}")
    print(f"[{elapsed()}] 时间顺序列存在情况：总 {len(time_cols)}，已存在 {len(exists_fields)}，新增 {len(missing_defs)}")
    
    # 创造临时表，然后从临时表中更新这两列
    _create_tmp_seq_table(engine, feeTblName, feeCoresMap)
    tmpTblName = 'TMP_TIMESEQ'
    deg = get_parallel_degree()
    update_sql = text(f"""
        MERGE /*+ parallel(f {deg}) parallel(t {deg}) use_hash(t) */ INTO {feeTblName} f
        USING {tmpTblName} t
        ON (f.ROWID = t.D0_ROWID)
        WHEN MATCHED THEN UPDATE SET
            f.D0_ROWID = t.item_time_seq
    """)
    with engine.begin() as conn:
        conn.execute(update_sql)
    print(f"[{elapsed()}] FEE 表更新 D0_ROWID, ITEM_TIME_SEQ 列完成")
    
    # 删除临时表
    drop_sql = text(f"DROP TABLE {tmpTblName} PURGE")
    with engine.begin() as conn:
        conn.execute(drop_sql)
    print(f"[{elapsed()}] 删除临时表 {tmpTblName} 完成")
    
    # 设置 D0_ROWID 为主键
    pk_name = f"PK_{feeTblName}_D0ROWID"
    pk_sql = text(f"""
        BEGIN
            EXECUTE IMMEDIATE 'ALTER TABLE {feeTblName} ADD CONSTRAINT {pk_name} PRIMARY KEY (D0_ROWID)';
        EXCEPTION
            WHEN OTHERS THEN
                NULL; -- 忽略所有错误，包括约束已存在的情况
        END;
    """)
    with engine.begin() as conn:
        conn.execute(pk_sql)
        print(f"[{elapsed()}] 设置主键 {pk_name} 完成")
        
# ------------------------------
#  增加其他索引
# ------------------------------
def prepare_fee_with_indexes(engine: Engine, feeTblName: str, feeCoresMap: Dict[str, str]):
    # 创建索引：setl_id, item_j_code, item_time, item_type, hsp_abbr, psn_no, item_std_code
    idx_cols: List[str] = []
    # 来自 core 的物理列
    for core_key in ['SETL_ID', 'ITEM_TIME', 'PSN_NO', 'ITEM_CODE', 'ITEM_MED_CODE', 'ITEM_HSP_CODE', 'ITEM_NAME', 'ITEM_HSP_NAME', 'VALI_FLAG']:
        idx_cols.append(feeCoresMap[core_key])
    # 派生/新增列（非 core）按物理名
    idx_cols.extend(['ITEM_J_CODE', 'ITEM_TYPE', 'HSP_ABBR', 'ITEM_STD_CODE'])
    # 去重保持顺序
    seen = set()
    uniq_cols = []
    for c in idx_cols:
        if c and c.upper() not in seen:
            uniq_cols.append(c)
            seen.add(c.upper())
    # 创建索引
    created = []
    for col in uniq_cols:
        if not _column_exists(engine, feeTblName, col):
            print(f"[{elapsed()}] 列不存在，跳过索引: {feeTblName}.{col}")
            continue
        safe_tbl = ''.join(ch for ch in feeTblName.upper() if ch.isalnum())[:8]
        safe_col = ''.join(ch for ch in col.upper() if ch.isalnum())[:18]
        idx_name = f"IDX_{safe_tbl}_{safe_col}"[:30]
        deg = get_parallel_degree()
        ddl = text(f"""
            BEGIN
                EXECUTE IMMEDIATE 'CREATE INDEX {idx_name} ON {feeTblName} ({col}) NOLOGGING PARALLEL {deg}';
            EXCEPTION
                WHEN OTHERS THEN
                    NULL; -- 忽略所有错误，包括索引已存在的情况
            END;
        """)
        with engine.begin() as conn:
            conn.execute(ddl)
        created.append(idx_name)
        print(f"[{elapsed()}] 创建索引 {idx_name} 完成")
    print(f"[{elapsed()}] 创建索引完成，共创建 {len(created)} 个索引: {', '.join(created)}")

# ------------------------------
#  处理冲减
# ------------------------------
def _add_combined_index(engine: Engine, table: str, index_name: str, columns: List[str]):
    # 创建联合索引 ITEM_TIME_SEQ, ITEM_J_CODE，先判断有没有
    ddl = text(f"""
        BEGIN
            EXECUTE IMMEDIATE 'CREATE INDEX {index_name} ON {table} ({", ".join(columns)}) NOLOGGING PARALLEL {get_parallel_degree()}';
        EXCEPTION
            WHEN OTHERS THEN
                NULL; -- 忽略所有错误，包括索引已存在的情况
        END;
    """)
    with engine.begin() as conn:
        conn.execute(ddl)
    print(f"[{elapsed()}] 创建联合索引 {index_name} 完成")

def _add_columns_for_reduction(engine: Engine, table: str):
    # 新增 J_Q, J_C, J_B 三列 number ; 新增 IS_J_REDUCE 列 varchar2(1)
    existing = _get_existing_columns(engine, table)
    needed = [
        ('J_Q', 'NUMBER'),
        ('J_C', 'NUMBER'),
        ('J_B', 'NUMBER'),
        ('IS_J_REDUCE', 'VARCHAR2(1)'),
    ]
    missing_defs: List[tuple[str,str]] = []
    exists_fields: List[str] = []
    for col, dtype in needed:
        if col.upper() in existing:
            exists_fields.append(col)
        else:
            missing_defs.append((col, dtype))
    if missing_defs:
        _add_columns(engine, table, missing_defs)
        print(f"[{elapsed()}] 批量新增冲减列 {len(missing_defs)} 个: {', '.join([c for c,_ in missing_defs])}")
    print(f"[{elapsed()}] 冲减列存在情况：总 {len(needed)}，已存在 {len(exists_fields)}，新增 {len(missing_defs)}")
    return exists_fields

def _update_reduction(engine: Engine, table: str, feeCoresMap: Dict[str, str]):
    # 计算出冲减标识以及冲减后的 J_Q, J_C, J_B
    setl_col = feeCoresMap.get('SETL_ID', 'SETL_ID')
    item_time_seq_col = 'D0_ROWID'
    q_col = feeCoresMap['Q_RAW']
    c_col = feeCoresMap['C_RAW']
    b_col = feeCoresMap['B_RAW']

    deg = get_parallel_degree()
    sql = text(f"""
        MERGE /*+ parallel(tgt {deg}) */ INTO {table} tgt
        USING (
            WITH ordered AS (
                SELECT
                    ROWID AS rid,
                    item_j_code,
                    {setl_col} AS setl_id,
                    {item_time_seq_col} AS item_time_seq,
                    {q_col} AS q_raw,
                    {c_col} AS c_raw,
                    {b_col} AS b_raw,
                    SUM({q_col}) OVER (
                        PARTITION BY {setl_col}, item_j_code
                        ORDER BY {item_time_seq_col}, ROWID
                        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
                    ) AS running_total,
                    SUM({q_col}) OVER (
                        PARTITION BY {setl_col}, item_j_code
                        ORDER BY {item_time_seq_col}, ROWID
                        ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
                    ) AS running_before
                FROM {table}
            ),
            staged AS (
                SELECT
                    ordered.*,
                    MIN(running_total) OVER (
                        PARTITION BY setl_id, item_j_code
                        ORDER BY item_time_seq, rid
                        ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING
                    ) AS min_future_running
                FROM ordered
            ),
            calc_q AS (
                SELECT
                    rid,
                    q_raw,
                    c_raw,
                    b_raw,
                    CASE
                        WHEN q_raw IS NULL THEN NULL
                        WHEN q_raw > 0 THEN GREATEST(min_future_running - NVL(running_before, 0), 0)
                        WHEN q_raw <= 0 THEN 0
                        ELSE q_raw
                    END AS new_j_q
                FROM staged
            ),
            final_calc AS (
                SELECT
                    rid,
                    new_j_q,
                    CASE
                        WHEN q_raw IS NULL THEN c_raw
                        WHEN q_raw <= 0 THEN c_raw
                        WHEN new_j_q >= q_raw THEN c_raw
                        WHEN new_j_q <= 0 THEN 0
                        WHEN c_raw IS NULL THEN NULL
                        ELSE c_raw * new_j_q / q_raw
                    END AS new_j_c,
                    CASE
                        WHEN q_raw IS NULL THEN b_raw
                        WHEN q_raw <= 0 THEN b_raw
                        WHEN new_j_q >= q_raw THEN b_raw
                        WHEN new_j_q <= 0 THEN 0
                        WHEN b_raw IS NULL THEN NULL
                        ELSE b_raw * new_j_q / q_raw
                    END AS new_j_b,
                    CASE
                        WHEN q_raw < 0 THEN '1'
                        WHEN q_raw > 0 AND new_j_q = 0 THEN '1'
                        ELSE NULL
                    END AS new_is_reduce
                FROM calc_q
            )
            SELECT rid, new_j_q, new_j_c, new_j_b, new_is_reduce
            FROM final_calc
        ) src
        ON (tgt.ROWID = src.rid)
        WHEN MATCHED THEN UPDATE SET
            tgt.J_Q = src.new_j_q,
            tgt.J_C = src.new_j_c,
            tgt.J_B = src.new_j_b,
            tgt.IS_J_REDUCE = src.new_is_reduce
    """)

    with engine.begin() as conn:
        res = conn.execute(sql)
        affected = res.rowcount if res.rowcount is not None else 0
    print(f"[{elapsed()}] 冲减计算完成，更新行数 {affected}")

def prepare_fee_with_reduction(engine: Engine, feeTblName: str, feeCoresMap: Dict[str, str]):
    # 为 FEE 表新增 冲减相关列，并创建联合索引
    # 1. 新增列
    _add_columns_for_reduction(engine, feeTblName)
    # 2. 创建联合索引
    idx_name = f"IDX_{feeTblName[:8]}_TIMESEQ_JCODE"[:30]
    _add_combined_index(engine, feeTblName, idx_name, ['ITEM_TIME_SEQ', 'ITEM_J_CODE'])
    # 3. 更新冲减标识与数值
    _update_reduction(engine, feeTblName, feeCoresMap)
    print(f"[{elapsed()}] 冲减处理完成")

def prepare_fee_with_btn(engine: Engine, feeTblName: str, feeCoresMap: Dict[str, str]):
    # 创建开关列 J_BTN
    # 1. 新增列
    str_col = 'J_BTN'
    if not _column_exists(engine, feeTblName, str_col):
        _add_column(engine, feeTblName, str_col, 'VARCHAR2(1)')
        print(f"[{elapsed()}] 新增列 {feeTblName}.{str_col} 完成")
    
    # 2. 更新列值，默认全部 '1'
    deg = get_parallel_degree()
    update_sql = text(f"""
        UPDATE /*+ parallel({feeTblName} {deg}) */ {feeTblName}
        SET {str_col} = '1'
    """)
    with engine.begin() as conn:
        res = conn.execute(update_sql)
        updated_cnt = res.rowcount or 0
    print(f"[{elapsed()}] 列 {feeTblName}.{str_col} 默认值更新完成，更新行数 {updated_cnt}")
    
    # 将不符合的关掉
    # 如果 J_B == 0，则关闭，'0'
    update_jb_sql = text(f"""
        UPDATE /*+ parallel({feeTblName} {deg}) */ {feeTblName}
        SET {str_col} = '0'
        WHERE J_B = 0
    """)
    with engine.begin() as conn:
        res = conn.execute(update_jb_sql)
        offed_cnt = res.rowcount or 0
    print(f"[{elapsed()}] 列 {feeTblName}.{str_col} 关闭非医保主单完成，关闭 '0' 的行数 {offed_cnt}")
    
    # 如果 vali_flag != '1'，则关闭，'0'
    vali_flag_col = feeCoresMap.get('VALI_FLAG', 'VALI_FLAG')
    update_vali_sql = text(f"""
        UPDATE /*+ parallel({feeTblName} {deg}) */ {feeTblName}
        SET {str_col} = '0'
        WHERE {vali_flag_col} IS NOT NULL AND {vali_flag_col} != '1'
    """)
    with engine.begin() as conn:
        res = conn.execute(update_vali_sql)
        offed_cnt_2 = res.rowcount or 0
    print(f"[{elapsed()}] 列 {feeTblName}.{str_col} 关闭无效数据完成，关闭 '0' 的行数 {offed_cnt_2}")
    
    # 如果 J_REDUCE = '1'，则关闭，'0'
    is_reduce_col = 'IS_J_REDUCE'
    update_reduce_sql = text(f"""
        UPDATE /*+ parallel({feeTblName} {deg}) */ {feeTblName}
        SET {str_col} = '0'
        WHERE {is_reduce_col} = '1'
    """)
    with engine.begin() as conn:
        res = conn.execute(update_reduce_sql)
        offed_cnt_3 = res.rowcount or 0
    print(f"[{elapsed()}] 列 {feeTblName}.{str_col} 关闭冲减数据完成，关闭 '0' 的行数 {offed_cnt_3}")
    
    # 如果在 m 表中查不到对应的 setl_id，则关闭，'0'
    setl_col = feeCoresMap.get('SETL_ID', 'SETL_ID')
    merge_sql = text(f"""
        UPDATE /*+ parallel({feeTblName} {deg}) */ {feeTblName} f
        SET {str_col} = '0'
        WHERE NOT EXISTS (
            SELECT 1 FROM m
            WHERE m.SETL_ID = f.{setl_col}
        )
        """)
    with engine.begin() as conn:
        res = conn.execute(merge_sql)
        offed_cnt_4 = res.rowcount or 0
    print(f"[{elapsed()}] 列 {feeTblName}.{str_col} 关闭无结算数据完成，关闭 '0' 的行数 {offed_cnt_4}")
    print(f"[{elapsed()}] 开关列 {feeTblName}.{str_col} 处理完成")
    
    # 针对 J_BTN 创建索引
    safe_tbl = ''.join(ch for ch in feeTblName.upper() if ch.isalnum())[:8]
    idx_name = f"IDX_{safe_tbl}_J_BTN"[:30]
    deg = get_parallel_degree()
    ddl = text(f"""
        BEGIN
            EXECUTE IMMEDIATE 'CREATE INDEX {idx_name} ON {feeTblName} ({str_col}) NOLOGGING PARALLEL {deg}';
        EXCEPTION
            WHEN OTHERS THEN
                NULL; -- 忽略所有错误，包括索引已存在的情况
        END;
    """)
    with engine.begin() as conn:
        conn.execute(ddl)
    print(f"[{elapsed()}] 创建索引 {idx_name} 完成")

    # 额外针对 ITEM_J_CODE, HSP_ABBR, FEE_OCUR_TIME（ITEM_TIME）, J_B 创建复合索引，加快 灌入 查询
    _add_combined_index(
        engine,
        feeTblName,
        "IDX_FEE_HSP_ABBR_ITEM_TIME_ITEM_J_CODE_J_B",
        ["ITEM_J_CODE", "HSP_ABBR", feeCoresMap['ITEM_TIME'], "J_BTN"]
    )
    
    # 统计、使得新鲜
    analyze_sql = text(f"BEGIN DBMS_STATS.GATHER_TABLE_STATS(ownname => '{DB_USER}', tabname => '{feeTblName.upper()}', cascade => TRUE); END;")
    with engine.begin() as conn:
        conn.execute(analyze_sql)
    print(f"[{elapsed()}] 复合索引创建完成")

# ------------------------------
#  创建视图
# ------------------------------
def prepare_fee_with_view(
        engine: Engine, feeTblName: str, setlTblName: str,
        feeCoresMap: Dict[str, str], setlCoresMap: Dict[str, str]
    ):
    view_name = 'D'

    # 找到需要的衍生列，这些列就在 fee 表当中
    # 'item_j_code', 'item_std_code', 'item_type', 'd0_rowid', 'item_day', 'item_hour'
    needed_fields_fee = [
        ('fee', 'ITEM_J_CODE', 'ITEM_J_CODE'),
        ('fee', 'ITEM_STD_CODE', 'ITEM_STD_CODE'),
        ('fee', 'ITEM_STD_NAME', 'ITEM_STD_NAME'),
        ('fee', 'ITEM_TYPE', 'ITEM_TYPE'),
        ('fee', 'D0_ROWID', 'D0_ROWID'),
        ('fee', 'ITEM_TIME_SEQ', 'D0_ROWID'),
        ('fee', 'ITEM_DAY', 'ITEM_DAY'),
        ('fee', 'ITEM_HOUR', 'ITEM_HOUR'),
    ]
    # 找出需要的 fee 表核心列
    needed_cores_fee = [
        'SETL_ID', 'ITEM_CODE', 'ITEM_MED_CODE', 'ITEM_HSP_CODE',
        'ITEM_NAME', 'ITEM_HSP_NAME', 'ITEM_TIME', 'ITEM_DEPT', 'APPLY_DEPT', 'P'
    ]
    # 找出需要的 fee 表的另外一些衍生列，大部分是来自于 setl 的，其中已经全部在 fee 表中新增
    needed_fields_setl = [
        ('fee', 'Q', 'J_Q'),
        ('fee', 'C', 'J_C'),
        ('fee', 'B', 'J_B'),
        ('fee', 'HSP_ABBR', 'HSP_ABBR'),
        ('fee', 'J_ISIN', 'J_ISIN'),
        ('fee', 'J_HAS_B', 'J_HAS_B'),
        ('fee', 'J_HSP_LV', 'J_HSP_LV'),
        ('fee', 'IN_DAY', 'IN_DAY'),
        ('fee', 'OUT_DAY', 'OUT_DAY'),
        ('fee', 'PERIOD', 'PERIOD'),
        ('fee', 'J_GEND', 'J_GEND'),
        ('fee', 'J_YEAROLD', 'J_YEAROLD'),
        ('fee', 'J_DAYOLD_IN', 'J_DAYOLD_IN'),
        ('fee', 'J_DAYOLD_OUT', 'J_DAYOLD_OUT'),
        ('fee', 'J_YEAR', 'J_YEAR'),
    ]
    # 找出需要的 setl 表核心列，这些列已经在 fee 表中新增
    needed_cores_setl = [
        'MDTRT_ID', 'PSN_NO', 'IN_DEPT', 'OUT_DEPT', 'IN_TIME', 'OUT_TIME', 'IN_DEPT', 'OUT_DEPT', 'AGE',
        'MED_TYPE', 'INSUTYPE',
    ]
    # 拼合成为最终的 field 列表
    needed_fields = needed_fields_fee
    needed_fields += [( 'fee', core, feeCoresMap[core] ) for core in needed_cores_fee if core in feeCoresMap]
    needed_fields += needed_fields_setl
    needed_fields += [( 'fee', core, setlCoresMap[core] ) for core in needed_cores_setl if core in setlCoresMap]

    # 创建这些列的 SELECT 片段
    selectSql = ',\n    '.join([
        f"{alias}.{col} AS {as_name}" for alias, as_name, col in needed_fields
    ])
    
    # 创建视图
    create_sql = text(f"""
        CREATE OR REPLACE VIEW {view_name} AS
        SELECT
            {selectSql},
            CAST(NULL AS VARCHAR2(400)) AS ITEM_J_NAME
        FROM {feeTblName} fee
        WHERE 1=1
            AND fee.J_BTN = '1'
    """)
    with engine.begin() as conn:
        conn.execute(create_sql)
    print(f"[{elapsed()}] 创建视图 {view_name} 完成")

# ------------------------------
#  主流程
# ------------------------------
def main():
    parser = argparse.ArgumentParser(description='step5_3_prepare_d_view: 可从指定步骤开始执行')
    parser.add_argument('-s', '--start-step', type=int, default=3, choices=[3,4,5,6,7,8], help='从第几步开始执行 (3-8)')
    args = parser.parse_args()
    start_step = args.start_step

    # 读取配置与环境
    struct_df = read_env_structure(STRUCTURE_FILE)
    
    setl_table, mdtrt_table, fee_table = resolve_table_names()
    
    # 获得 setl 表与 mdtrt 表的核心列映射
    setlCoresMap = resolve_cores(struct_df, 'SETL')
    mdtrtCoresMap = resolve_cores(struct_df, 'MDTRT')
    feeCoresMap = resolve_cores(struct_df, 'FEE')

    # 创建数据库连接
    engine = create_db_engine()
    # 开启并行能力
    setup_oracle_parallel_session(engine, get_parallel_degree())

    print(f"[{elapsed()}] 从第 {start_step} 步开始执行")
    if start_step <= 3:
        print(f"[{elapsed()}] Step 3: 新增 SETL 派生字段")
        prepare_fee_with_setl(engine, fee_table, setl_table, feeCoresMap, setlCoresMap)
    if start_step <= 4:
        print(f"[{elapsed()}] Step 4: 日期计算 + 标准编码/名称/类别 计算")
        prepare_fee_with_dates(engine, fee_table, feeCoresMap)
        prepare_fee_with_std_code(engine, fee_table, feeCoresMap)
    if start_step <= 5:
        print(f"[{elapsed()}] Step 5: 生成 J 项目编码")
        prepare_fee_with_j_code(engine, fee_table, feeCoresMap)
    if start_step <= 6:
        print(f"[{elapsed()}] Step 6: 生成时间顺序唯一码并设置主键/唯一索引/其他索引")
        prepare_fee_with_time_seq(engine, fee_table, feeCoresMap)
        prepare_fee_with_indexes(engine, fee_table, feeCoresMap)
    if start_step <= 7:
        print(f"[{elapsed()}] Step 7: 处理冲减")
        prepare_fee_with_reduction(engine, fee_table, feeCoresMap)
        prepare_fee_with_btn(engine, fee_table, feeCoresMap)
    if start_step <= 8:
        print(f"[{elapsed()}] Step 8: 创建视图 D")
        prepare_fee_with_view(engine, fee_table, setl_table, feeCoresMap, setlCoresMap)

if __name__ == '__main__':
    main()
    print(f'[{elapsed()}] 主流程完成') 